package com.haopt.elasticjob.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.dataflow.DataflowJob;
import com.haopt.elasticjob.annotation.ElasticSimpleJob;
import com.haopt.elasticjob.config.MyJobShardingStrategy;
import com.haopt.elasticjob.config.MyListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * DataflowJob流式任务案例
 * 模拟10000个数据
 *
 * @author haopt
 */
@Slf4j
@Component
@ElasticSimpleJob(jobName = "myDataflowJob",
        corn = "0/7 * * * * ?",
        shardingTotalCount = 10,
        overwrite = true,
        monitorExecution = true,
        failover = true,
        jobShardingStrategyClass = MyJobShardingStrategy.class,
        listeners = MyListener.class
)
public class DataFlowJobTest implements DataflowJob<Integer> {
    /**
     * 奇数
     */
    private static List<Integer> DATA_ODD;
    /**
     * 偶数
     */
    private static List<Integer> DATA_EVENT;

    static {
        DATA_ODD = new ArrayList<>();
        DATA_EVENT = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            if (i % 2 == 0) {
                DATA_EVENT.add(i);
            } else {
                DATA_ODD.add(i);
            }
        }
    }

    @Override
    public List<Integer> fetchData(ShardingContext shardingContext) {
        int item = shardingContext.getShardingItem();
        List<Integer> subList = null;
        //判断如果是0号分片，从偶数集合中每次抓取100个数据
        if (item == 0) {
            if (DATA_EVENT.size() >= 100) {
                subList = DATA_EVENT.subList(0, 100);
            } else {
                subList = DATA_EVENT.subList(0, DATA_EVENT.size());
            }
            //判断如果是1号分片，从奇数集合中每次抓取100个数据
        } else if (item == 1) {
            if (DATA_ODD.size() >= 100) {
                subList = DATA_ODD.subList(0, 100);
            } else {
                subList = DATA_ODD.subList(0, DATA_ODD.size());
            }
        }
        return subList;
    }

    @Override
    public void processData(ShardingContext shardingContext, List<Integer> list) {
        int item = shardingContext.getShardingItem();
        log.info(item + "号分片，处理的数据" + list);
        if (item == 0) {
            DATA_EVENT.removeAll(list);
        } else if (item == 1) {
            DATA_ODD.removeAll(list);
        }
    }
}
